package com.wxj.bank.mq.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListeners;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.io.*;
import java.util.HashMap;
import java.util.Map;

/**
 * @author shengtao
 * @Description: rabbitmq配置类
 * @date 2019/02/01 18:23
 */

@Configuration
@ConfigurationProperties(prefix = "bank.rabbitmq")
public class RabbitMqConfig {

    private String host;

    private String username;

    private String password;

    private String virtualHost;

    private int cacheSize;

    private int channelCacheSize;

    /* exchange */
    private String exchange;
    private String rowkey;
    private String queue;

    /* 延迟队列 */
    private String originalQueue;

    private String originalExchange;

    private String originalRowkey;

    private String deadQueue;

    private String deadExchange;

    private String deadRowkey;

    @Bean
    public Binding deadBinding(){
        AmqpAdmin amqpAdmin = amqpAdmin();
        Queue deaQueue = new Queue(deadQueue,true);
        amqpAdmin.declareQueue(deaQueue);
        DirectExchange deaeExchage = new DirectExchange(deadExchange,true,false);
        amqpAdmin.declareExchange(deaeExchage);
        Binding binding = BindingBuilder.bind(deaQueue).to(deaeExchage).with(deadRowkey);
        amqpAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    public Binding originBinding(){
      AmqpAdmin amqpAdmin = amqpAdmin();
      Map<String, Object> arguments = new HashMap<>();
      arguments.put("x-dead-letter-exchange",deadExchange);
      arguments.put("x-dead-letter-routing-key",deadRowkey);
      Queue oriQueue = new Queue(originalQueue,true,false,false,arguments);
      //队列设置过期事件
      amqpAdmin.declareQueue(oriQueue);
      DirectExchange oriExchage = new DirectExchange(originalExchange,true,false);
      amqpAdmin.declareExchange(oriExchage);
      Binding binding = BindingBuilder.bind(oriQueue).to(oriExchage).with(originalRowkey);
      amqpAdmin.declareBinding(binding);
      return binding;
    }


    @Bean
    Queue queue(){
        return new Queue(queue,Boolean.TRUE);
    }

    @Bean
    DirectExchange exchange(){
        return new DirectExchange(exchange,Boolean.TRUE,Boolean.FALSE);
    }

    @Bean
    Binding binding(){
        return BindingBuilder.bind(queue()).to(exchange()).with(rowkey);
    }

    @Bean
    Queue fanoutQueue(){
        return new Queue("fanout.queue1",true);
    }
    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout.exchange1",true,false);
    }
    @Bean
    Binding fanoutBinding(){
        return BindingBuilder.bind(fanoutQueue()).to(fanoutExchange());
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        connectionFactory.setChannelCacheSize(channelCacheSize);
        connectionFactory.setConnectionCacheSize(cacheSize);
        connectionFactory.setConnectionThreadFactory(new CustomizableThreadFactory("rabbitmq-"));
        connectionFactory.setVirtualHost(virtualHost);
//        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        return connectionFactory;
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(false);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setBackOffPolicy(new ExponentialBackOffPolicy());
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy());
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(500);
        backOffPolicy.setMultiplier(10.0);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        template.setRetryTemplate(retryTemplate);
        return template;
    }

    @Bean
    public RabbitListenerContainerFactory<DirectMessageListenerContainer> rabbitListenerContainerFactory(){
        DirectRabbitListenerContainerFactory containerRabbitListenerContainerFactory = new DirectRabbitListenerContainerFactory();
        containerRabbitListenerContainerFactory.setConnectionFactory(connectionFactory());
        containerRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        containerRabbitListenerContainerFactory.setConsumersPerQueue(3);
        /*containerRabbitListenerContainerFactory.setMessageConverter(new MessageConverter() {
            @Override
            public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
                ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                try {
                    ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
                    objectOutputStream.writeObject(o);
                    byte[] content = outputStream.toByteArray();
                    Message message = new Message(content,messageProperties);
                    return message;
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return null;
            }

            @Override
            public Object fromMessage(Message message) throws MessageConversionException {
                byte[] content = message.getBody();
                ByteArrayInputStream inputStream = new ByteArrayInputStream(content);
                try {
                    ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
                    Object o = objectInputStream.readObject();
                    System.out.println(o);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
                MessageProperties properties = message.getMessageProperties();
                return message;
            }
        });*/
        return containerRabbitListenerContainerFactory;
    }

    /**
     * 配置queue、rowkey、exchange
     */



    public String getHost() {
        return host;
    }
    public void setHost(String host) {
        this.host = host;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getVirtualHost() {
        return virtualHost;
    }

    public void setVirtualHost(String virtualHost) {
        this.virtualHost = virtualHost;
    }

    public int getCacheSize() {
        return cacheSize;
    }

    public void setCacheSize(int cacheSize) {
        this.cacheSize = cacheSize;
    }

    public int getChannelCacheSize() {
        return channelCacheSize;
    }

    public void setChannelCacheSize(int channelCacheSize) {
        this.channelCacheSize = channelCacheSize;
    }

    public String getExchange() {
        return exchange;
    }

    public void setExchange(String exchange) {
        this.exchange = exchange;
    }

    public String getRowkey() {
        return rowkey;
    }

    public void setRowkey(String rowkey) {
        this.rowkey = rowkey;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }

    public String getOriginalQueue() {
        return originalQueue;
    }

    public void setOriginalQueue(String originalQueue) {
        this.originalQueue = originalQueue;
    }

    public String getOriginalExchange() {
        return originalExchange;
    }

    public void setOriginalExchange(String originalExchange) {
        this.originalExchange = originalExchange;
    }

    public String getOriginalRowkey() {
        return originalRowkey;
    }

    public void setOriginalRowkey(String originalRowkey) {
        this.originalRowkey = originalRowkey;
    }

    public String getDeadQueue() {
        return deadQueue;
    }

    public void setDeadQueue(String deadQueue) {
        this.deadQueue = deadQueue;
    }

    public String getDeadExchange() {
        return deadExchange;
    }

    public void setDeadExchange(String deadExchange) {
        this.deadExchange = deadExchange;
    }

    public String getDeadRowkey() {
        return deadRowkey;
    }

    public void setDeadRowkey(String deadRowkey) {
        this.deadRowkey = deadRowkey;
    }
}
